package com.cnpc.bds.sdk.sink.kafka.executor;

import com.cnpc.bds.sdk.sink.kafka.handler.Pipeline;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 后置任务流执行线程.
 *
 * @author ssp
 * @since 1.0
 */
@Slf4j
public class EventLoop implements Runnable {

    private final BlockingQueue<Object> eventQueue = new LinkedBlockingQueue<>();

    private Pipeline pipeline;

    public EventLoop(Pipeline pipeline) {
        this.pipeline = pipeline;
    }

    public void fireEventAsync(Object msg) {
        eventQueue.offer(msg);
    }

    @Override
    public void run() {
        while (true) {
            try {
                final Object msg = eventQueue.take();
                pipeline.firePipeline(msg); //同步执行

            } catch (Exception e) {
                log.error("Do event loop exception!", e);
            }
        }
    }


    public static EventLoop newOne(Pipeline pipeline) {
        return new EventLoop(pipeline);
    }

}
